package com.ww.disruptor.consumer;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.ww.disruptor.event.ProductEvent;
import com.ww.disruptor.factory.CustomizableThreadFactory;
import com.ww.disruptor.factory.ProductEventFactory;
import com.ww.disruptor.handler.ProductEventHandler;
import com.ww.disruptor.producer.ProductEventProducer;

import java.util.UUID;
import java.util.concurrent.ThreadFactory;

public class ConsumerDemo01 {

    public static void main(String[] args) {
//        test01();

//        test03();

        test04();
    }

    /**
     * 单生产者单消费者
     */
    public static void test01() {
        ThreadFactory threadFactory = new CustomizableThreadFactory("customer-pool-");

        // 单生产者模式，指定ProducerType.SINGLE，性能会快一点
        Disruptor<ProductEvent> disruptor = new Disruptor<>(new ProductEventFactory(), 4, threadFactory, ProducerType.SINGLE, new YieldingWaitStrategy());

        // 单消费者模式
        disruptor.handleEventsWith(new ProductEventHandler());

        // 启动
        disruptor.start();

        RingBuffer<ProductEvent> ringBuffer = disruptor.getRingBuffer();
        ProductEventProducer producer = new ProductEventProducer(ringBuffer);

        for (int i = 0; i < 6; i++) {
            producer.onData(i + "," + UUID.randomUUID());
        }

        // 终止
        disruptor.shutdown();
    }

    /**
     * 单生产者多消费者广播模式（广播模式）
     */
    public static void test02() {
        ThreadFactory threadFactory = new CustomizableThreadFactory("customer-pool-");
        Disruptor<ProductEvent> disruptor = new Disruptor<>(new ProductEventFactory(), 4, threadFactory, ProducerType.SINGLE, new YieldingWaitStrategy());

        // 多消费者广播模式
        EventHandler<ProductEvent>[] cusomers = new EventHandler[4];
        for (int i = 0; i < cusomers.length; i++) {
            cusomers[i] = new ProductEventHandler();
        }

        disruptor.handleEventsWith(cusomers);

        // 启动
        disruptor.start();

        RingBuffer<ProductEvent> ringBuffer = disruptor.getRingBuffer();
        ProductEventProducer producer = new ProductEventProducer(ringBuffer);
        for (int i = 0; i < 6; i++) {
            producer.onData(i + "," + UUID.randomUUID());
        }

        // 终止
        disruptor.shutdown();
    }

    /**
     * 单生产者多消费者广播模式（集群模式）
     * 使用WorkPool必须保证Consumer对应一个线程，否则当RingBuffer满的时候，Producer和Consumer都会阻塞，因此Disruptor废弃了使用Executor的构造方法
     * ExecutorService executorService = new ThreadPoolExecutor(2, 5, 1, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(5, true), new CustomizableThreadFactory("customer-pool-"));
     * 推荐使用的是提供ThreadFactory形式的构造器，后续会根据事件处理器的个数来新增对应的线程
     */
    public static void test03() {
        ThreadFactory threadFactory = new CustomizableThreadFactory("customer-pool-");
        Disruptor<ProductEvent> disruptor = new Disruptor<>(new ProductEventFactory(), 4, threadFactory, ProducerType.SINGLE, new YieldingWaitStrategy());

        // 多消费者广播模式
        WorkHandler<ProductEvent>[] cusomers = new ProductEventHandler[4];
        for (int i = 0; i < cusomers.length; i++) {
            cusomers[i] = new ProductEventHandler();
        }

        disruptor.handleEventsWithWorkerPool(cusomers);

        // 启动
        disruptor.start();

        RingBuffer<ProductEvent> ringBuffer = disruptor.getRingBuffer();
        ProductEventProducer producer = new ProductEventProducer(ringBuffer);
        for (int i = 0; i < 6; i++) {
            producer.onData(i + "," + UUID.randomUUID());
        }

        // 终止
        disruptor.shutdown();
    }

    /**
     * 多生产者多消费者广播模式（集群模式）
     */
    public static void test04() {
        ThreadFactory threadFactory = new CustomizableThreadFactory("customer-pool-");
        Disruptor<ProductEvent> disruptor = new Disruptor<>(new ProductEventFactory(), 4, threadFactory, ProducerType.MULTI, new YieldingWaitStrategy());

        // 多消费者广播模式
        WorkHandler<ProductEvent>[] cusomers = new ProductEventHandler[4];
        for (int i = 0; i < cusomers.length; i++) {
            cusomers[i] = new ProductEventHandler();
        }

        disruptor.handleEventsWithWorkerPool(cusomers);

        // 启动
        disruptor.start();

        RingBuffer<ProductEvent> ringBuffer = disruptor.getRingBuffer();
        ProductEventProducer producer1 = new ProductEventProducer(ringBuffer);
        ProductEventProducer producer2 = new ProductEventProducer(ringBuffer);
        for (int i = 0; i < 6; i++) {
            producer1.onData(20 + i + "," + UUID.randomUUID());
            producer2.onData(10 + i + "," + UUID.randomUUID());
        }

        // 终止
        disruptor.shutdown();
    }
}
